/**
 * Copyright 2008-2009. Chongqing Communications Industry Services Co.,Ltd Information Technology Branch. All rights
 * reserved. <a>http://www.cqcis.com</a>
 */
package com.cqcis.uip.base.framework.utils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.cqcis.uip.base.framework.core.LogService;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor.LogThreadPoolExecutor;
import com.cqcis.uip.base.logadaptor.LogConfiguration;

/**
 * 循环执行的任务的线程池
 * 
 * @author huangmy
 * @date   2008-3-8
 */
public class LoopThreadPoolExecutor implements Executor {
	private static Log log = LogFactory.getLog(LoopThreadPoolExecutor.class);
    private long intervalTime; // 出错后间隔时间
    private Map<Runnable, Object> waitLocks = new ConcurrentHashMap<Runnable, Object>();
    private LogThreadPoolExecutor tpe; // 线程池    
    private volatile boolean shutdowned = false;
    
    private LoopThreadPoolExecutor() {    	
    }
    
    /**
     * 唤醒所有池中的等待线程
     * 
     */
    public void notifyPool() {
    	for (Object lock : this.waitLocks.values()) {
    		synchronized (lock) {
        		lock.notifyAll();
    		}
    	}    	    	
    }
    
    /**
     * 带日志服务的循环线程池
     * 
     * @param coreSize
     * @param maxSize
     * @param keepAliveTime
     * @param logService
     * @param logConfiguration
     * @param threadPoolName
     * @param intervalTime
     * @return
     */
    public static LoopThreadPoolExecutor getExecutor(int coreSize, int maxSize, int keepAliveTime, 
            final LogService logService, final LogConfiguration logConfiguration, final String threadPoolName, long intervalTime) {
    	
    	LoopThreadPoolExecutor ltpe = new LoopThreadPoolExecutor((LogThreadPoolExecutor)LogServiceThreadPoolExecutor.getLogExecutor(coreSize, maxSize, keepAliveTime, logService, logConfiguration, threadPoolName));
    	ltpe.setIntervalTime(intervalTime);
    	
    	return ltpe;
    }
    
    /**
     * 带日志服务的循环线程池
     * 
     * @param coreSize
     * @param maxSize
     * @param keepAliveTime
     * @param serviceContext
     * @param threadPoolName
     * @param intervalTime
     * @return
     */
    public static LoopThreadPoolExecutor getExecutor(int coreSize, int maxSize, int keepAliveTime, 
            RegisteredServiceContext serviceContext, final String threadPoolName, long intervalTime) {
    	return getExecutor(coreSize, maxSize, keepAliveTime, serviceContext.getLogService(), 
    			getLogConfiguration(serviceContext), threadPoolName, intervalTime);
    }
    
    private static LogConfiguration getLogConfiguration(RegisteredServiceContext serviceContext) {
		RegisteredService rs = serviceContext.getRegisteredService();
		if (rs != null) {
			return rs.getLogConfiguration();
		}
		return null;
	}
    
    /**
     * 不带日志服务的循环线程池
     * 
     * @param coreSize
     * @param maxSize
     * @param keepAliveTime
     * @param intervalTime
     * @return
     */
    public static Executor getExecutor(int coreSize, int maxSize, int keepAliveTime, long intervalTime) {
    	return getExecutor(coreSize, maxSize, keepAliveTime, null, null, null, intervalTime);
    }
    
    private LoopThreadPoolExecutor(LogThreadPoolExecutor tpe) {
    	this.tpe = tpe;
    }
    
    public int getPoolSize() {
    	return this.tpe.getPoolSize();
    }
        		
	public void execute(final Runnable command) {
		Runnable runnAble = new Runnable() {
			public void run() {
				Object lock = new Object();
				waitLocks.put(this, lock);
				while (true) {
					// 检查执行该任务的线程是否已经退出
					if (shutdowned) {
						break;
					}
					
					try {
						command.run();
						
						try {
							synchronized (lock) {
								lock.wait(intervalTime);
							}
						} catch (Throwable e) {							
						}						
					} catch (Throwable t) {
						log.info("Catch Throwable in loop.", t);
						try {
							synchronized (lock) {
								lock.wait(intervalTime);
							}
						} catch (Throwable e) {							
						}
					}
				}
			}
		};
		
		tpe.execute(runnAble);
	}
	
	public void shutdown() {
		shutdowned = true;
		if (tpe != null) {
			try {
				for (int i = 0; i < 10; i ++) {
					tpe.shutdown();
					tpe.shutdownWait(2000);
				}
				
				tpe.shutdown();
				tpe.shutdownWait(2000);
				tpe.shutdown();
			    tpe.shutdownWait();			    
			} catch (Throwable t) {				
			}
		}
	}

	public long getIntervalTime() {
		return intervalTime;
	}

	public void setIntervalTime(long intervalTime) {
		this.intervalTime = intervalTime;
	}
}
